fix: revert unsafe partial aggregates after final fallback - #5421

cc @andygrove @comphead discovered this correctness issue when running TPC-DS benchmark internally with spark.comet.shuffle.enabled=false

I used an LLM (Claude Code) to help with this review, including building both PRs locally and running the sweeps below. I have gone over the results myself and I agree with them, but flagging the tooling up front.

Thanks for digging into this. The gap you identified is real, and the reasoning about canAggregateBeConverted skipping the child-native check matches what the comment in operators.scala says the tagging pass is supposed to cover. Before getting into the mechanics I want to raise something about the framing.

I swept every aggregate function through the Comet-Partial plus Spark-Final configuration, using spark.comet.exec.shuffle.enabled=false so the Final stays in Spark, over 8 rows in 4 Parquet files with a filter that leaves three partials empty. On Spark 4.1 with JDK 17, on current main, only two of the 26 aggregates I tried are wrong across that boundary:

avg(bigint)          comet=[null]                      spark=[1.0]
avg(decimal(20,2))   comet=threw ARITHMETIC_OVERFLOW   spark=[200.000000]

With this PR applied the decimal case is fixed and AVG(bigint) is still wrong, exactly as before. CometAverage.supportsMixedPartialFinal returns true for non-decimal input, so the allowlist waves through what looks like the same defect with the same root cause. That seems like the thing to fix first, and it would be worth a test either way, since it is the same bug this PR is named after.

Everything else the PR newly blocks was already correct in my sweep. COUNT, decimal SUM, FIRST, LAST, and the whole stddev/variance/covariance/correlation family all matched Spark on main, and all of them lose their native Partial with this change. I tried to justify that cost by hunting for a case where the guard earns it: decimal SUM overflowing DECIMAL(38,0), long SUM overflow, TRY_SUM, TRY_AVG, grouped and filtered variants, under ANSI on and off. All of those matched with the native Partial retained, including ANSI throwing and legacy returning null. The comment on CometSum.supportsMixedPartialFinal about overflow detection not surviving the split did not reproduce in this direction, which makes sense to me because that comment describes the Spark-partial to Comet-final direction rather than this one. The same goes for the COUNT exclusion, which the docstring attributes to PropagateEmptyRelationAfterAQE and Spark 4.0 count-bug decorrelation. Both of those only bite when the Final becomes a CometHashAggregateExec.

The other thing I noticed is that #5420 already contains revertUnsafePartialAggregates verbatim, so this PR is a subset of it. When I keep only #5420's native avg.rs and avg_decimal.rs changes and drop the planner guard entirely, both AVG cases are fixed and every aggregate keeps its native Partial.

So on the evidence I have, the native change in #5420 fixes strictly more than this guard does and costs no native aggregation, while this guard fixes less and gives up native partials for a fair number of aggregates. Could you say what this PR adds once #5420 lands? The description asserts the safeguard is still needed after that, but I was not able to construct a case, and none of the four new tests exercises one. If the value is defense in depth against buffer mismatches we have not found yet, rather than a bug that is broken today, that seems like a reasonable position to me, but I think it should be argued on that basis and weighed against the measured cost. It also affects the ordering of the two PRs.

Two smaller things while I was in there. supportsMixedPartialFinal is a single direction-agnostic flag, but most of the exclusions behind it are justified only in the Spark-partial to Comet-final direction. Would it be worth splitting it into two predicates so this pass consults only the one that applies? And where revertChain returns None, the unsafe boundary is left in the plan with no signal, which is the same failure mode the PR exists to prevent. A plain warning there would be too noisy, since I see the None path taken benignly on q10 and q35 where the Partial is already Spark, but gating it on findCometPartialAgg finding a Comet Partial that revertChain failed to reach would make the gap between those two traversals detectable.

One last note, unrelated to this PR. The sweep also turned up that PERCENTILE, APPROX_PERCENTILE, COLLECT_LIST and COLLECT_SET all trip the strict check with spark.comet.exec.shuffle.enabled=false, reporting "Comet did not convert ObjectHashAggregate but recorded no fallback reason". The shuffle-enabled guard on ObjectHashAggregateExec declines without calling withFallbackReason. I will file that separately.